"""Visual analyzer (M9): VLM-powered actor/symbols detection with frame sampling.

- Responsibilities:
  - Sample frames from a video path (config-driven caps).
  - Build prompt using `config/categories.yaml` for actor and symbols groups.
  - Call VLM API with timeout + retries; parse JSON result.
  - Count frequencies and select the single highest-frequency item for `actor` and `symbols`.
  - Error isolation: on failure, return `{"actor": "NA", "symbols": "NA"}`.

- Rules: keys from env only, logging, PEP8, type annotations.
"""
from __future__ import annotations

import json
import logging
import os
import tempfile
from typing import Any, Dict, List, Optional, Tuple

import yaml  # type: ignore
import requests  # type: ignore

from src.utils import config_loader

LOGGER = logging.getLogger(__name__)


def _load_categories(categories_path: str = "config/categories.yaml") -> Dict[str, Any]:
    try:
        with open(categories_path, "r", encoding="utf-8") as f:
            data = yaml.safe_load(f) or {}
        return data or {}
    except Exception as exc:
        LOGGER.error("invalid_format: categories yaml load failed: %s", exc)
        return {}


def _pick_cap(cfg: Dict[str, Any]) -> int:
    # Use frame sampling config, but cap globally to avoid heavy IO during unit tests.
    fs = config_loader.get_frame_sampling_config(cfg)
    return int(fs.get("global_max_frames", 4))


def _sample_video_frames(video_path: str, num_frames: int) -> List[str]:
    # Simplified sampling: for unit tests we mock this; here we just create temp placeholders.
    # In production, integrate FFmpeg to extract evenly spaced frames.
    out: List[str] = []
    try:
        for i in range(num_frames):
            fd, tmp_path = tempfile.mkstemp(prefix=f"vlm_frame_{i}_", suffix=".png")
            os.close(fd)
            out.append(tmp_path)
        return out
    except Exception as exc:
        LOGGER.error("processing_failed: frame sampling failed: %s", exc)
        return []


def _collect_options(categories: Dict[str, Any], key: str) -> List[str]:
    for cat in categories.get("categories", []):
        if cat.get("key") == key:
            opts = cat.get("options")
            res: List[str] = []
            if isinstance(opts, list):
                for o in opts:
                    if isinstance(o, dict):
                        name = o.get("name") or o.get("key")
                        if name:
                            res.append(str(name))
                    elif isinstance(o, str):
                        res.append(o)
            return res
    return []


def _collect_symbols_groups(categories: Dict[str, Any]) -> Dict[str, List[str]]:
    for cat in categories.get("categories", []):
        if cat.get("key") == "symbols":
            groups = cat.get("groups", [])
            res: Dict[str, List[str]] = {}
            for g in groups:
                key = str(g.get("key"))
                opts: List[str] = []
                for o in g.get("options", []) or []:
                    if isinstance(o, dict):
                        name = o.get("name") or o.get("key")
                        if name:
                            opts.append(str(name))
                    elif isinstance(o, str):
                        opts.append(o)
                res[key] = opts
            return res
    return {}


def _build_vlm_prompt(categories: Dict[str, Any]) -> str:
    actor_opts = ", ".join(_collect_options(categories, "actor"))
    sym_groups = _collect_symbols_groups(categories)
    group_lines = []
    for gk, opts in sym_groups.items():
        if opts:
            group_lines.append(f"{gk}: {', '.join(opts)}")
    parts = [
        "请分析以下视频帧，识别并返回JSON，且仅返回JSON：",
        "字段：actors(列表) 与 symbols(列表)。",
        "actors 候选：" + actor_opts if actor_opts else "actors 候选：不限",
        "symbols 按分组（scene/tool/language/music）给出候选：",
        *group_lines,
        "请按每一帧识别一个或多个项，最终返回所有帧的合并结果列表，不要解释。",
    ]
    return "\n".join(parts)


def _call_vlm(frames: List[str], prompt: str, api_conf: Dict[str, Any]) -> str:
    base_url = api_conf.get("base_url", "https://api.example.com")
    timeout = int(api_conf.get("timeout", 30))
    access_key = os.environ.get("VLM_ACCESS_KEY", "")
    secret_key = os.environ.get("VLM_SECRET_KEY", "")
    if not access_key or not secret_key:
        raise RuntimeError("missing_api_key: VLM keys not set")
    # Example schema; provider-dependent. Tests will mock this function.
    url = base_url.rstrip("/") + "/v1/analyze"
    headers = {
        "X-Access-Key": access_key,
        "X-Secret-Key": secret_key,
        "Content-Type": "application/json",
    }
    payload = {
        "prompt": prompt,
        "frames": frames,  # In production, frames would be URLs or base64 images
    }
    resp = requests.post(url, headers=headers, json=payload, timeout=timeout)
    resp.raise_for_status()
    return resp.text


def _count_top(items: List[str]) -> Tuple[str, Dict[str, int]]:
    counts: Dict[str, int] = {}
    for it in items:
        if not it:
            continue
        k = str(it)
        counts[k] = counts.get(k, 0) + 1
    if not counts:
        return "NA", counts
    # Deterministic tie-break: by (-count, name)
    top = sorted(counts.items(), key=lambda kv: (-kv[1], kv[0]))[0][0]
    return top, counts


def _parse_vlm_result(text: str) -> Tuple[List[str], List[str]]:
    obj = json.loads(text)
    actors = obj.get("actors") or []
    symbols = obj.get("symbols") or []
    if not isinstance(actors, list) or not isinstance(symbols, list):
        raise RuntimeError("invalid_format: VLM result must contain lists")
    return [str(a) for a in actors], [str(s) for s in symbols]


def analyze_video_content(video_path: str, cfg: Dict[str, Any]) -> Dict[str, str]:
    """Analyze visual content to produce highest-frequency actor and symbols.

    Returns dict: {"actor": str, "symbols": str}
    """
    categories_path = cfg.get("paths", {}).get("categories_yaml", "config/categories.yaml")
    categories = _load_categories(categories_path)

    cap = _pick_cap(cfg)
    frames = _sample_video_frames(video_path, cap)
    if not frames:
        LOGGER.warning("file_not_found or empty_frames: video=%s", video_path)
        return {"actor": "NA", "symbols": "NA"}

    prompt = _build_vlm_prompt(categories)
    api_conf = cfg.get("apis", {}).get("vlm", {})

    try:
        text = _call_vlm(frames, prompt, api_conf)
        actors, symbols = _parse_vlm_result(text)
        top_actor, _ = _count_top(actors)
        top_symbol, _ = _count_top(symbols)
        return {"actor": top_actor, "symbols": top_symbol}
    except Exception as exc:
        LOGGER.warning("vlm_fallback_NA: %s", exc)
        return {"actor": "NA", "symbols": "NA"}
